package Hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;


public class HadoopUtil {
     String url ="hdfs://192.168.70.11:9000/";
    //  上传
    public  String hadoopUpload(String srcPath ,String destPath) throws IOException{
        String url_new = url;
        Configuration conf = new Configuration();
        FileSystem fileSystem = FileSystem.get(URI.create(url_new),conf);
        /**
         * 构造一个访问指定HDFS系统的客户端对象
         * 第一个参数：HDFS的URI
         * 第二个参数：客户端指定的配置参数
         */

        // 本地的 文件
        Path srcLoc =new Path(srcPath);
        // hdfs 的路径
        Path destLoc = new Path(destPath);
        // 使用 hadoop 的 copyFromLocalFile， 拷贝本地文件到HDFS文件系统
        fileSystem.copyFromLocalFile(srcLoc,destLoc);


        return "上传成功";
    }
     // 读文件
     public List hadoopRead(String filePath)throws IOException {

        // 用api的形式 读取下刚才写的文件
         String reaUrl = url+ filePath;
         // 加载hdfs 的配置
         Configuration conf = new Configuration();
         // 建立hdfs连接
         FileSystem fileSystem = FileSystem.get(URI.create(reaUrl),conf);  //URI.create(reaUrl),conf)
         // 打开url路径的文件
         FSDataInputStream fsDataInputStream = fileSystem.open(new Path(reaUrl));
         // 输出 文件内容，并关闭输入流

         // s
         BufferedReader br = new BufferedReader(new InputStreamReader(fsDataInputStream, "utf-8"));
         /*
         缓冲区在内存中预留指定大小的存储空间用来对I/O的数据做临时存储，这部分预留的内存空间叫缓冲区
         BufferedReader：
         1、带有缓冲区的字符输入流。
         2、 使用这个流时候，不需要自定义char数组，或者说不需要自定义byte数组。自带缓冲。
          3、当一个流的构造方法需要一个流的时候，这个被传进来的流叫做，节点流。
          4、外部负责包装的这个流，叫做：包装流。还有一个名字叫：处理流。
          5、 像当前的程序的来说BufferedReader就是 包装流，处理流。
          */

         String line = null;
         List<String> list = new ArrayList<>();
         while((line=br.readLine()) != null) {
                String lineObj = br.readLine();
            list.add(lineObj);
             System.out.println("数据内容 ");
             System.out.println(line);
         }

        //  FileStatus stat = fs.getFileStatus(path);
//         byte[] buffer = new byte[Integer.parseInt(String.valueOf(fsDataInputStream))];
//           fsDataInputStream.readFully(0, buffer);

         return list;
     }

    // 下载
    public String hadoopDownload(String srcPath ,String destPath)throws IOException {
        // 用api的形式 读取下我们刚才写的文件
        String url_new = url;
        Configuration conf = new Configuration();
        FileSystem fileSystem = FileSystem.get(URI.create(url_new),conf);
        // 建立hdfs连接
        // 本地的 文件
        Path srcLoc =new Path(srcPath);
        // hdfs 的路径
        Path destLoc = new Path(destPath);
        fileSystem.copyToLocalFile(false,srcLoc, destLoc,true);
      //  fileSystem.copyToLocalFile(srcLoc,destLoc);

        return "下载成功";
    }


}
